package com.cli.springboot_rabbitmq.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.HashMap;

/**
 * @Author xiongmin
 * @Description
 * @Date 2021/2/26 10:51
 * @Version 1.0
 **/
@Component
@Slf4j
@RabbitListener(queues = {"fanout.A", "fanout.B"}) //监听的队列名称 fanout.A
//@RabbitListener(queues = "fanout.A") //监听的队列名称 fanout.A
public class FanoutConsumerListener {

    @RabbitHandler
    public void receive(@Payload HashMap msg, Channel channel, Message message) {
        try {
            log.info("FanoutReceiver消费者收到消息  : ");
            if (null != msg && !msg.isEmpty()) {
                msg.forEach((key, value) -> {
                    log.info(key + ": " + value + "\n");
                });
            } else {
                log.info("msg is empty");
            }
            // 确认消息被消费
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
            try {
                if (message.getMessageProperties().getRedelivered()) {
                    log.error("消息已重复处理失败,拒绝再次接收...");
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false); // 拒绝消息
                } else  {
                    log.error("消息即将返回队列再次处理");
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
                }
            } catch (Exception ex) {
                log.error("消息消费异常时处理失败",ex.getMessage(),ex);
            }

        }
    }
}
